package com.loren.demo07.cluster.subscriber;

import com.alibaba.fastjson.JSONObject;
import com.loren.demo07.cluster.dto.ClusterMessage;
import com.loren.demo07.cluster.websocket.container.ClusterClientContainer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@Slf4j
@Component
public class ClusterMessageSubscriber {

    public static final String CHANNEL_NAME = "websocket-channel";

    /**
     * 发布订阅模式 - 订阅
     */
    public ClusterMessageSubscriber(RedisConnectionFactory redisConnectionFactory) {
        RedisConnection redisConnection = redisConnectionFactory.getConnection();
        redisConnection.subscribe((message, pattern) -> {
            ClusterMessage socketMessage = JSONObject.parseObject(new String(message.getBody(), StandardCharsets.UTF_8), ClusterMessage.class);
            try {
                if (StringUtils.isBlank(socketMessage.getToClientId())) {
                    this.sendTextMessage(socketMessage.getMessage());
                } else {
                    this.sendTextMessage(socketMessage.getToClientId(), socketMessage.getMessage());
                }
            } catch (IOException e) {
                log.error("an error has occured, message = {}", e.getMessage());
            }

        }, CHANNEL_NAME.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 发送给所有客户端
     */
    public void sendTextMessage(String message) throws IOException {
        for (Map.Entry<String, WebSocketSession> entry : ClusterClientContainer.ONLINE_CLIENTS.entrySet()) {
            this.sendTextMessage(entry.getKey(), message);
        }
    }

    /**
     * 发送给指定客户端
     */
    public void sendTextMessage(String clientId, String message) throws IOException {
        if (ClusterClientContainer.ONLINE_CLIENTS.containsKey(clientId)) {
            ClusterClientContainer.ONLINE_CLIENTS.get(clientId).sendMessage(new TextMessage(message.getBytes()));
            log.info("send a message to clientId = {}, message = {}", clientId, message);
        } else {
            log.warn("cannot find the connection for clientId = {}", clientId);
        }
    }
}
